package com.business.etl.flow.mq;

import cn.hutool.core.util.IdUtil;
import cn.hutool.core.util.StrUtil;
import com.alibaba.fastjson.JSON;
import com.component.api.model.flow.FlowInfo;
import com.business.etl.flow.service.FlowService;
import com.framework.rocketmq.consumer.Consumer;
import com.framework.rocketmq.consumer.MQConfig;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.common.message.MessageExt;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.nio.charset.StandardCharsets;

/**
 * 描述：
 * 流程自定义消费者
 * @author xianggj
 * @Date 2021/10/14 16:14
 **/
@Slf4j
@Component
public class FlowConsumer extends Consumer {

    private static final String TOPIC = "flow";



    @Autowired
    private FlowService flowService;


    @Override
    protected void consume(MessageExt msg, ConsumeConcurrentlyContext context) throws Exception {
        String content = new String(msg.getBody(), StandardCharsets.UTF_8);
        log.info("msg:" + content);
        FlowInfo flowInfo= null;
        try {
            flowInfo = JSON.parseObject(content, FlowInfo.class);
            //添加流程唯一标识 一般会传 不然流程无法追踪
            if (StrUtil.isBlank(flowInfo.getFlowId())){
                flowInfo.setFlowId(IdUtil.fastSimpleUUID());
            }
        } catch (Exception e) {
            e.printStackTrace();
            log.error("无效消息！"+ content);
            return;
        }
        flowService.readyRun(flowInfo);
    }

    @Override
    protected MQConfig getMQConfig() {
        return new MQConfig(TOPIC);
    }

    @Override
    protected String getGroupName() {
        //消费实例
        return "flowGroupName";
    }
}
